package com.github.xiaolyuh.manager.impl;

import com.alibaba.fastjson.JSON;
import com.github.xiaolyuh.base.BasicCacheManager;
import com.github.xiaolyuh.configcenter.ConfigCenter;
import com.github.xiaolyuh.constant.CacheConstant;
import com.github.xiaolyuh.dataloader.DataLoader;
import com.github.xiaolyuh.entity.CacheKey;
import com.github.xiaolyuh.entity.CacheLockObject;
import com.github.xiaolyuh.entity.CacheValue;
import com.github.xiaolyuh.enums.CacheCodeEnum;
import com.github.xiaolyuh.exception.CacheException;
import com.github.xiaolyuh.lock.CacheLock;
import com.github.xiaolyuh.manager.ICacheManager;
import com.github.xiaolyuh.util.StringUtils;
import com.github.xiaolyuh.utils.FastJSONHelper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Predicate;

/**
 * 二级缓存管理类
 * <p>
 * 一级建议使用机器的缓存
 * 二级建议使用分布式缓存
 *
 * @author xuwanglong
 */
public class CacheManagerImpl implements ICacheManager {

    private static final Logger logger = LoggerFactory.getLogger(CacheManagerImpl.class);

    private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 10, 0L,
            TimeUnit.SECONDS, new ArrayBlockingQueue(100000), new ThreadFactoryBuilder().setNameFormat("cache-refresh-pool-%d").build());

    @Override
    public <T, K> T get(CacheKey<K> cacheKey, Class<T> clazz, DataLoader<K, T> dataLoader, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        //check args
        checkArgs(cacheKey, clazz);

        String key = cacheKey.getKey();
        // 未开启缓存，直接走业务逻辑
        if (!isLocalCacheEnable() && !isDistributeEnable()) {
            if(dataLoader == null){
                return null;
            }
            CacheValue<T> cacheItem = dataLoader.get(cacheKey.getId());
            if (null == cacheItem) {
                return null;
            }
            return cacheItem.getValue();
        }

        try {
            Boolean setLocalCache = false;
            Object value = null;
            // 本地缓中获取
            if (canLocalCacheEnable(localCacheManager)) {
                value = localCacheManager.get(key, clazz);
            }
            // 需要设置本地缓存
            if (value == null) {
                setLocalCache = true;
            }

            // redis缓存中获取
            if (value == null && canDistributeEnable(distributeCacheManager)) {
                value = distributeCacheManager.get(key, clazz);
            }
            // 缓存有对应的数据，直接响应
            if (null != value) {
                if (isFake(value)) {
                    return null;
                }
                if (setLocalCache && canLocalCacheEnable(localCacheManager)) {
                    localCacheManager.set(key, value, CacheConstant.CACHE_DEFAULT_TIMEOUT);
                }
                return (T)value;
            }
            if(dataLoader == null){
                // 不回源，直接返回空
                return null;
            }else{
                return reload(cacheKey, clazz, dataLoader, distributeCacheManager, localCacheManager, key);
            }

        } catch (Exception e) {
            logger.error(String.format("[cache]get value error, key:[%s]", FastJSONHelper.serialize(cacheKey)), e);
            throw new CacheException(CacheCodeEnum.FAIL.getCode(), e.getMessage());
        } catch (Throwable throwable) {
            logger.error(String.format("[cache]get value error, key:[%s]", FastJSONHelper.serialize(cacheKey)), throwable);
            throw new CacheException(CacheCodeEnum.FAIL);
        }
    }

    private <T, K> T reload(CacheKey<K> cacheKey, Class<T> clazz, DataLoader<K, T> dataLoader, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager, String key) throws CacheException{
        // 通过业务加载方式操作
        CacheLockObject lock;
        if (CacheLock.getInstance().isLocked(key)) {
            lock = CacheLock.getInstance().getLock(key);
        } else {
            // 加锁
            CacheLock.getInstance().tryLock(key);
            lock = CacheLock.getInstance().getLock(key);
        }
        // currentLock == null这中场景极少发生，如果发生则是：线程1检查时发现有其他线程持有了锁，但到这步骤时，对应线程已经释放了锁
        // 此时其实已经触发了业务逻辑操作,那缓存对应的数据是可信的
        if (lock == null || lock.isOptFlagFinish()) {
            return getFromCache(key, clazz, distributeCacheManager, localCacheManager);
        }

        synchronized (lock) {
            // 保证只有一个线程并发操作业务加载逻辑
            if (!lock.isOptFlagFinish()) {
                try {
                    CacheValue<T> cacheValue = this.reload(cacheKey, dataLoader, distributeCacheManager, localCacheManager);
                    if(cacheValue == null){
                        return null;
                    }else{
                        return cacheValue.getValue();
                    }
                } finally {
                    CacheLock.getInstance().unLock(key);
                    // 乐观：无论是否发生异常，都认为本次操作OK，因后续的还会触发加载操作
                    lock.setOptFlagFinish(true);
                }
            }
        }
        return getFromCache(key, clazz, distributeCacheManager, localCacheManager);
    }

    @Override
    public <T, K> void refresh(final CacheKey<K> cacheKey, final Class<T> clazz, final DataLoader<K, T> dataLoader, boolean async, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        // 检查参数
        checkArgs(cacheKey, clazz);
        // 参数检验
        String key = cacheKey.getKey();
        // 未开启缓存，直接走业务逻辑
        if (!isDistributeEnable() && !isLocalCacheEnable()) {
            return;
        }
        try {
            final String lockKey = "refresh_" + key;
            if (CacheLock.getInstance().isLocked(lockKey)) {
                return;
            }
            // 尝试加锁
            if (!CacheLock.getInstance().tryLock(lockKey)) {
                return;
            }
            // 保证只有一个线程并发操作业务加载逻辑
            try {
                if (async) {
                    Runnable task = newTask(cacheKey, dataLoader, distributeCacheManager, localCacheManager);
                    submitTask(task, cacheKey.getKey());
                } else {
                    reload(cacheKey, dataLoader, distributeCacheManager, localCacheManager);
                }
            } finally {
                CacheLock.getInstance().unLock(lockKey);
            }
        } catch (Exception e) {
            logger.error(String.format("[cache]refresh value error, key:[%s]", key), e);
            throw new CacheException(CacheCodeEnum.FAIL.getCode(), e.getMessage());
        } catch (Throwable throwable) {
            logger.error(String.format("[cache]refresh value error, key:[%s]", key), throwable);
            throw new CacheException(CacheCodeEnum.FAIL);
        }
    }

    @Override
    public boolean delete(String key, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        try {
            boolean localResult = true;
            boolean distrResult = true;
            if (canLocalCacheEnable(localCacheManager)) {
                localResult = localCacheManager.delete(key);
            }
            if (canDistributeEnable(distributeCacheManager)) {
                distrResult = distributeCacheManager.delete(key);
            }
            return localResult && distrResult;
        } catch (Throwable e) {
            logger.error(String.format("[cache]delete value error, key:[%s]", key), e);
            throw new CacheException(CacheCodeEnum.FAIL);
        }
    }

    private <K, T> Runnable newTask(final CacheKey<K> cacheKey, final DataLoader<K, T> dataLoader, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) {
        return () -> {
            try {
                reload(cacheKey, dataLoader, distributeCacheManager, localCacheManager);
            } catch (Throwable e) {
                logger.warn("[cache]async reload cache error, cacheKey=" + JSON.toJSONString(cacheKey), e);
            }
        };
    }

    private <T, K> CacheValue<T> reload(final CacheKey<K> cacheKey, final DataLoader<K, T> dataLoader, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        CacheValue<T> cacehValue = dataLoader.get(cacheKey.getId());
        if (null == cacehValue || null == cacehValue.getValue()) {
            this.setFake(cacheKey, distributeCacheManager, localCacheManager);
            return null;
        } else {
            // 更新对应的缓存，可能发生异常
            this.set(cacheKey, cacehValue, distributeCacheManager, localCacheManager);
        }
        return cacehValue;
    }

    private void submitTask(Runnable task, String cacheKey) {
        try {
            threadPool.submit(task);
        } catch (Exception e) {
            logger.warn("[cache]async reload cache thread pool error, cacheKey=" + cacheKey, e);
        }
    }

//    private void setMissLocalCache(List<String> keyList, BatchGetCache redisVo, BasicCacheManager localCacheManager) throws CacheException {
//        if (!canLocalCacheEnable(localCacheManager)) {
//            return;
//        }
//        if (redisVo != null && redisVo.getResultMap() != null && keyList != null) {
//            for (String key : keyList) {
//                Object value = redisVo.getResultMap().get(key);
//                if (value == null || isFake(value)) {
//                    continue;
//                }
//                localCacheManager.set(key, value, CacheConstant.CACHE_DEFAULT_TIMEOUT);
//            }
//        }
//    }

    private <K, T> Map<K, T> mGetMap(Map<String, K> key2IdMap, Map<String, Object> resultMap, Class<T> clazz) {
        if (null == resultMap || resultMap.isEmpty()) {
            return Collections.emptyMap();
        }
        Map<K, T> rstMap = new HashMap();
        for (Map.Entry<String, Object> ele : resultMap.entrySet()) {
            String key = ele.getKey();
            Object value = ele.getValue();
            try {
                if (null == value || isFake(value)) {
                    continue;
                }
                K id = key2IdMap.get(key);
                if (null == id) {
                    continue;
                }
                rstMap.put(id, (T)value);
            } catch (Exception e) {
                logger.error(String.format("[cache]deserialize value error, key:[%s], value:[%s]", key, FastJSONHelper.serialize(value)), e);
            }
        }
        return rstMap;
    }

    private <T, K> Map<K, T> convertCacheValueMap(Map<K, CacheValue<T>> tmpMap) {
        if (null == tmpMap || tmpMap.isEmpty()) {
            return Collections.emptyMap();
        }
        Map<K, T> rstMap = new HashMap<>(tmpMap.size());
        for (Map.Entry<K, CacheValue<T>> obj : tmpMap.entrySet()) {
            K key = obj.getKey();
            CacheValue<T> val = obj.getValue();
            if (null == key || null == val || null == val.getValue()) {
                continue;
            }
            rstMap.put(key, val.getValue());
        }
        return rstMap;
    }

    private <T> T getFromCache(String key, Class<T> clazz, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        Object value = null;
        try {
            if (canLocalCacheEnable(localCacheManager)) {
                value = localCacheManager.get(key, clazz);
            }
            if (value == null && canDistributeEnable(distributeCacheManager)) {
                value = distributeCacheManager.get(key, clazz);
            }
            if (null == value || isFake(value)) {
                return null;
            }
            return (T)value;
        } catch (Exception e) {
            logger.error(String.format("[cache]get value error, key:[%s], value:[%s]", key, FastJSONHelper.serialize(value)), e);
            throw new CacheException(CacheCodeEnum.FAIL.getCode(), e.getMessage());
        } catch (Throwable throwable) {
            logger.error(String.format("[cache]get value error, key:[%s], value:[%s]", key, FastJSONHelper.serialize(value)), throwable);
            throw new CacheException(CacheCodeEnum.FAIL);
        }
    }

    @Override
    public <K, T> void set(CacheKey<K> cacheKey, CacheValue<T> cacheValue, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        checkArgs(cacheKey, cacheValue);
        String key = cacheKey.getKey();
        try {
            Object value = cacheValue.getValue();
            if (canLocalCacheEnable(localCacheManager)) {
                localCacheManager.set(key, value, cacheValue.getExpireTime());
            }
            if (canDistributeEnable(distributeCacheManager)) {
                distributeCacheManager.set(key, value, cacheValue.getExpireTime());
            }
        } catch (Exception e) {
            logger.error(String.format("[cache]set value error, key:[%s]", FastJSONHelper.serialize(cacheKey)), e);
            throw new CacheException(CacheCodeEnum.FAIL.getCode(), e.getMessage());
        } catch (Throwable throwable) {
            logger.error(String.format("[cache]set value error, key:[%s]", FastJSONHelper.serialize(cacheKey)), throwable);
            throw new CacheException(CacheCodeEnum.FAIL);
        }
    }

    @Override
    public <K, T> void setWithFake(CacheKey<K> cacheKey, CacheValue<T> cacheValue, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        checkArgsWithFake(cacheKey, cacheValue);

        if(cacheValue.getValue() == null){
            setFake(cacheKey, distributeCacheManager, localCacheManager);
        }else{
            set(cacheKey, cacheValue, distributeCacheManager, localCacheManager);
        }
    }

    private <K> void setFake(CacheKey<K> cacheKey, BasicCacheManager distributeCacheManager, BasicCacheManager localCacheManager) throws CacheException {
        try {
            String key = cacheKey.getKey();
            if (StringUtils.isBlank(key)) {
                return;
            }
            //本地缓存不设置fake值,避免redis中的fake值更新与本地的fake值不一致
            //注意:fake值会直接回源redis,但不会回源DB
            /*
            if (canLocalCacheEnable(localCacheManager)) {
                localCacheManager.set(key, CacheConstant.NIL_BYTES, CacheConstant.CACHE_DEFAULT_TIMEOUT);
            }*/
            if (canDistributeEnable(distributeCacheManager)) {
                distributeCacheManager.set(key, CacheConstant.NIL_STRING, CacheConstant.CACHE_DEFAULT_TIMEOUT);
            }
        } catch (Exception e) {
            logger.error(String.format("[cache]set fake value error, key:[%s]", FastJSONHelper.serialize(cacheKey)), e);
            throw new CacheException(CacheCodeEnum.FAIL.getCode(), e.getMessage());
        } catch (Throwable throwable) {
            logger.error(String.format("[cache]set fake value error, key:[%s]", FastJSONHelper.serialize(cacheKey)), throwable);
            throw new CacheException(CacheCodeEnum.FAIL);
        }
    }

    private boolean isFake(Object value) {
        if (null == value) {
            return false;
        }
        return Objects.equals(value, CacheConstant.NIL_STRING);
    }

    //**********************参数校验**********************

    private <T, K> void checkArgs(CacheKey cacheKey, Class<T> clazz) throws CacheException {
        if (null == cacheKey || null == cacheKey.getKey() || null == clazz) {
            throw new CacheException(CacheCodeEnum.PARAM_INVALID.getCode(), "cacheKey or class or cacheLoader is null");
        }
    }

    private <T, K> void checkArgs(CacheKey<K> cacheKey, CacheValue<T> cacheValue) throws CacheException {
        if (null == cacheKey || null == cacheValue || null == cacheKey.getKey() || null == cacheKey.getId() || null == cacheValue.getValue()) {
            throw new CacheException(CacheCodeEnum.PARAM_INVALID.getCode(), "cacheKey or cacheValue is null");
        }
    }

    private <T, K> void checkArgs(List<CacheKey<K>> cacheKeyList, Class<T> clazz) throws CacheException {
        if (null == cacheKeyList || null == clazz) {
            throw new CacheException(CacheCodeEnum.PARAM_INVALID.getCode(), "class or cacheKeyList or cacheValue is null");
        }
    }

    private <K, T> void checkArgsWithFake(CacheKey<K> cacheKey, CacheValue<T> cacheValue) throws CacheException {
        if (null == cacheKey || null == cacheValue || null == cacheKey.getKey()) {
            throw new CacheException(CacheCodeEnum.PARAM_INVALID.getCode(), "cacheKey or cacheValue is null");
        }
    }

    private static<T> Predicate<T> distinctByKey(Function<? super T, Object> keyExtractor) {
        Map<Object, Boolean> seen = new HashMap<>();
        return object -> seen.putIfAbsent(keyExtractor.apply(object), Boolean.TRUE) == null;
    }

    //**********************配置开关**********************

    protected Boolean isDistributeEnable() {
        return ConfigCenter.allowRedisCache();
    }

    protected Boolean isLocalCacheEnable() {
        return ConfigCenter.allowLocalCache();
    }

    private boolean canDistributeEnable(BasicCacheManager distributeCacheManager) {
        return isDistributeEnable() && Objects.nonNull(distributeCacheManager);
    }

    private boolean canLocalCacheEnable(BasicCacheManager localCacheManager) {
        return isLocalCacheEnable() && Objects.nonNull(localCacheManager);
    }

}
